package hadoop.mapreduce.reducerjoin;

import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.HashMap;

/**
 * 在reduce端合并表
 *
 *  一个分区对应一个ReduceTask,如果不指定分区,就一个分区.如果不写分区规则指定了分区个数,就使用默认分区规则
 *  ReduceTask与MapTask串行,将mt的结果分组,每个分组执行一次reduce方法
 */
public class TableReducer extends Reducer<Text, TableBean,TableBean, NullWritable> {


    /**
     * @param key pid
     * @param values 根据pid分组后的TableBean集合
     * @param context
     * @throws IOException
     * @throws InterruptedException
     *
     * values:
     * 1001	 01	  1  order
     * 1004	01	  4  order
     * 01  小米  pd
     */
    @Override
    protected void reduce(Text key, Iterable<TableBean> values, Context context) throws IOException, InterruptedException {
        ArrayList<TableBean> list=new ArrayList();
        HashMap<String,String> map=new HashMap();
        //由于Hadoop对内存用到极致,values中的TableBean对象不是new的,会发生覆盖,所以要自己创建对象接受values中的数据
        for (TableBean value : values) {
            if ("order".equals(value.getFlag())){
                TableBean tableBean = new TableBean();
                try {
                    BeanUtils.copyProperties(tableBean,value);
                } catch (IllegalAccessException e) {
                    e.printStackTrace();
                } catch (InvocationTargetException e) {
                    e.printStackTrace();
                }
                list.add(tableBean);
            }else {
                //map中key存商品id value存商品名称
                map.put(value.getPid(),value.getpName());
            }
        }

        for (TableBean tableBean : list) {
            tableBean.setpName(map.get(tableBean.getPid()));
            //写出时调用tableBean的toString()方法
            context.write(tableBean,NullWritable.get());
        }

    }
}
